Feature/composite workflow execution v1 by stevanbz · Pull Request #1 · stevanbz/alerting

stevanbz · 2023-02-13T18:55:05Z

Issue #, if available:

Description of changes:

Created WorkflowRunner logic that iterates through the list of the monitors and sequentially executes monitor depending of the type
Created logic for getting the chained monitor finding doc ids (two steps) -> 1. getting the finding docIds per execution (using the workflowRunContext) 2. Based on the docIds used in the first step relevant documents are determined for the currently processed monitor (Relevant classes: WorkflowService, CompositeWorkflowRunner, WorkflowRunContext - which is instantiated every time run happens)
Created TransportExecuteWorkflowAction and RestExecuteWorkflowAction
Added integration tests that are testing the workflow execution. WorkflowRunnerIT.test execute workflow with custom alerts and finding index with doc level and bucket level delegates can be ignored because in AlertingSingleNode test cases it's very hard to create bucket level monitors since this test suite is mocking ScriptService which is responsible for loading Scripts

CheckList:
[ ] Commits are signed per the DCO using --signoff

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

…enario Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep · 2023-02-20T19:07:41Z

can you look into adding this painless script module at plugin load test
there will be some method you can override and register the necessary plugin

Let's look into how we can verify composite monitors containing bucket level monitors

…nitor index is not initialized yet. Added workflow crud test cases Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep

Thanks for the changes, Stevan
have reviewed 50% of the PR
will review more while you can address the comments

eirsep · 2023-02-23T19:00:10Z

alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt

 import org.opensearch.action.ActionRequest
 import org.opensearch.action.ActionResponse
 import org.opensearch.alerting.action.ExecuteMonitorAction
+import org.opensearch.alerting.action.ExecuteWorkflowAction


let's discuss offline about cluster/node level settings for composite workflows

Ok sounds good.

eirsep · 2023-02-23T19:20:27Z

alerting/src/main/kotlin/org/opensearch/alerting/BucketLevelMonitorRunner.kt

        periodEnd: Instant,
-        dryrun: Boolean
+        dryrun: Boolean,
+        workflowExecutionContext: WorkflowRunContext?


NIT: name the variable same as the type

eirsep · 2023-02-23T23:07:01Z

alerting/src/main/kotlin/org/opensearch/alerting/BucketLevelMonitorRunner.kt

                            else QueryBuilders.boolQuery().must(source.query())
                            queryBuilder.filter(QueryBuilders.termsQuery(fieldName, bucketValues))
+
+                            if (workflowRunContext != null && !workflowRunContext.indexToDocIds.isNullOrEmpty()) {


why are we applying this logic here and not in InputService where the actual search query is being executed ?

Yeah you are right. This logic can be removed from here - I forgot to remove it once I added in input service. Tnx and good catch!

eirsep · 2023-02-23T23:20:28Z

alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt

            }

+            // If monitor execution is triggered from a workflow
+            val indexToRelatedDocIdsMap = workflowRunContext?.indexToDocIds


can we intialize this just before its usage instead of here?

eirsep · 2023-02-24T06:55:49Z

alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt

        }
    }

+    private fun updateInputQueryWithFindingDocIds(


add comments/javadocs to explain what we intend to do wherever we are using chained findings filtering

eirsep · 2023-02-24T06:56:31Z

alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt

+
+                        // Rewrite query to consider the doc ids per given index
+                        if (chainedFindingExist(indexToDocIds)) {
+                            val updatedSourceQuery = updateInputQueryWithFindingDocIds(input.query.query(), indexToDocIds!!)


null check for query required?

why are we changing at input query
add a filter after search query is constructed

null check for query required?

You are right. Adding the null check.

why are we changing at input query add a filter after search query is constructed

Since rewrittenQuery.query() returns QueryBuilder()! (which can be null) we must do a cast to a BoolQueryBuilder (I guess) which then later we would need to set again to a rewrittenQuery.query.
You can see here that later on query is transformed in a String so it wouldn't be so straight forward to add a filter.
I don't have any more idea how to do this in elegant way (maybe lacking domain knowledge around the OpenSearch classes I can use for this purpose)- if you can give me a hint or a code snippet how I can do, it would be good. Tnx!

eirsep · 2023-02-24T07:09:25Z

alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt

        }
    }

+    private fun updateInputQueryWithFindingDocIds(


this should be a common methods used in all the monitor types

I thought the same - but then I saw that the search query is executed on a different way depending of the monitor type.
Ie. here you can see how the doc level monitor is getting the matching docs. So, for example, doc level monitor iterates through the list of indices and getting the documents index by index. That's why I adjusted getting the matched docIds on doc level monitor to be aligned with existing logic. Check it out here

eirsep · 2023-02-24T07:10:08Z

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

+    val xContentRegistry: NamedXContentRegistry,
+) {
+
+    suspend fun getFindingDocIdsPerMonitorExecution(chainedMonitor: Monitor, workflowExecutionId: String): Map<String, List<String>> {


getFindingDocIdsByExecutionId*

eirsep · 2023-02-24T07:30:03Z

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

+                    .seqNoAndPrimaryTerm(true)
+            )
+            .indices(chainedMonitor.dataSources.findingsIndex)
+        val searchResponse: SearchResponse = client.suspendUntil { client.search(searchRequest, it) }


handle indexNotFound

eirsep · 2023-02-24T07:32:40Z

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

+        return buildMonitors(searchResponse)
+    }
+
+    private fun buildMonitors(response: SearchResponse): List<Monitor> {


should this function be called parseMonitors

eirsep · 2023-02-24T07:33:55Z

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

+        return monitors
+    }
+
+    suspend fun getDocIdsPerFindingIndex(monitorId: String, workflowExecutionId: String): Map<String, List<String>> {


This function will be removed since it's not used at all.

…e workflow Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

… consider the workflow execution id Added worfklow service used for retrieving monitors and their findings. Added business logic for considering the chained monitors Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

…when loading the cluster Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep · 2023-02-24T07:34:39Z

alerting/src/main/kotlin/org/opensearch/alerting/action/ExecuteWorkflowRequest.kt

+
+class ExecuteWorkflowRequest : ActionRequest {
+    val dryrun: Boolean
+    val requestEnd: TimeValue


what is requestEnd?

Copy-paste of ExecuteMonitorRequest. Used in CompositeWorkflowRunner - and passed to concrete monitor runner - ie in bucketLevelMonitors used for defining the search params when creating findings. Check it out here

eirsep · 2023-02-24T07:34:47Z

alerting/src/main/kotlin/org/opensearch/alerting/action/ExecuteWorkflowRequest.kt

+import org.opensearch.commons.alerting.model.Workflow
+import java.io.IOException
+
+class ExecuteWorkflowRequest : ActionRequest {


javadocs for field

eirsep · 2023-02-24T07:35:07Z

alerting/src/main/kotlin/org/opensearch/alerting/action/ExecuteWorkflowRequest.kt

+    )
+
+    override fun validate(): ActionRequestValidationException? {
+        return null


validations?

Added check. Tnx and good point

eirsep · 2023-02-24T18:06:26Z

alerting/src/main/kotlin/org/opensearch/alerting/action/ExecuteWorkflowResponse.kt

+
+class ExecuteWorkflowResponse : ActionResponse, ToXContentObject {
+
+    val workflowRunResult: List<MonitorRunResult<*>>


we should store other fields like workflow execution start, and end time, status=failed, successful

Makes sense. Will add those fields and appropriate logic around them

eirsep · 2023-02-24T18:09:10Z

alerting/src/main/kotlin/org/opensearch/alerting/resthandler/RestExecuteWorkflowAction.kt

+        return listOf()
+    }
+
+    override fun replacedRoutes(): MutableList<RestHandler.ReplacedRoute> {


why do we need replacedRoutes. we are not replacing routes. this would be a new API

Removing class completely. Sorry

eirsep · 2023-02-24T23:53:21Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

-        TODO("Not yet implemented")
-    }
+    ): List<MonitorRunResult<*>> {
+        val workflowExecutionId = UUID.randomUUID().toString()


we should make this execution id more deterministic..
workflowId+timestamp

Something like:
val workflowExecutionId = UUID.randomUUID().toString() + LocalDateTime.now()
What do you think?

Changed to something like:
val executionId = workflow.id.plus(LocalDateTime.now()).plus(UUID.randomUUID().toString())

eirsep · 2023-02-27T18:25:39Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

-    ): MonitorRunResult<*> {
-        TODO("Not yet implemented")
-    }
+    ): List<MonitorRunResult<*>> {


This should return workflowRunResult which should contain list of monitorRunResult

eirsep · 2023-02-27T18:26:29Z

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

+        return indexToRelatedDocIdsMap
+    }
+
+    suspend fun searchMonitors(monitors: List<String>, size: Int, owner: String?): List<Monitor> {


rename to getMonitorsById

is owner field used?

No will remove it. Good catch

eirsep · 2023-02-27T18:27:06Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

+
+        val delegates = (workflow.inputs[0] as CompositeInput).sequence.delegates.sortedBy { it.order }
+        // Fetch monitors by ids
+        val monitors = monitorCtx.workflowService!!.searchMonitors(delegates.map { it.monitorId }, delegates.size, workflow.owner)


why do we need owner field?

We don't. Removing

eirsep · 2023-02-27T18:27:37Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

+        // Validate the monitors size
+        if (delegates.size != monitors.size) {
+            val diffMonitorIds = delegates.map { it.monitorId }.minus(monitors.map { it.id }.toSet()).joinToString()
+            throw IllegalStateException("Delegate monitors don't exist $diffMonitorIds")


Plz also log workflow id in the message

Will do. Also will add a logs on the beginning and end of workflow execution

…esponse class Code adjusted according to comments Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep · 2023-03-01T19:01:55Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

+                var indexToDocIds = mapOf<String, List<String>>()
+                var delegateMonitor: Monitor
+                delegateMonitor = monitorsById[delegate.monitorId]
+                    ?: throw IllegalStateException("Delegate monitor not found ${delegate.monitorId} for the workflow $workflow.id")


wrap with alerting exception

eirsep · 2023-03-01T19:05:41Z

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

+     * @param chainedMonitor Monitor that is previously executed
+     * @param workflowExecutionId Execution id of the current workflow
+     */
+    suspend fun getFindingDocIdsByExecutionId(chainedMonitor: Monitor, workflowExecutionId: String): Map<String, List<String>> {


handle indexNotFound and return empty

I was thinking and let me elaborate a little bit my thinking and proposed solution:
Let's catch all the exceptions that can be raised, and wrap them up in AlertingException (check it out here). The caller function - the function in CompositeWorkflowRunner (here) will do a check and return empty workflow run result. What do you think?

eirsep · 2023-03-01T19:23:12Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

+                    ?: throw IllegalStateException("Delegate monitor not found ${delegate.monitorId} for the workflow $workflow.id")
+                if (delegate.chainedFindings != null) {
+                    val chainedMonitor = monitorsById[delegate.chainedFindings!!.monitorId]
+                        ?: throw IllegalStateException("Chained finding monitor not found ${delegate.monitorId} for the workflow $workflow.id")


wrap with alerting exception

eirsep · 2023-03-01T19:46:56Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

+                        dryRun,
+                        workflowRunContext
+                    )
+                } else {


NIT: use else if for query level and throw unsupported exception

Should I also wrap into alerting exception or? Ie. something like this:
Something like this:
else if(delegateMonitor.isQueryLevelMonitor()){ QueryLevelMonitorRunner.runMonitor( delegateMonitor, monitorCtx, periodStart, periodEnd, dryRun, workflowRunContext ) } else { throw AlertingException.wrap( IllegalStateException("Unsupported monitor type") ) }

eirsep · 2023-03-01T19:47:56Z

alerting/src/main/kotlin/org/opensearch/alerting/workflow/WorkflowRunContext.kt

+data class WorkflowRunContext(
+    val chainedMonitorId: String?,
+    val workflowExecutionId: String,
+    val indexToDocIds: Map<String, List<String>>


indexToDocIds is not a good variable name. Someone reading the code would not understand that this is the input source.

What about "matchingDocIdsPerIndex"?

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep · 2023-03-08T19:21:54Z

Let's have latestRunTime and latestExecutionId in workflow object or workflow metadata object.

…dation if the query monitor is part of the workflow chain Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

lezzago · 2023-03-08T23:46:41Z

alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt

+        @JvmField val WORKFLOW_BASE_URI = "/_plugins/_alerting/workflows"
        @JvmField val DESTINATION_BASE_URI = "/_plugins/_alerting/destinations"
        @JvmField val LEGACY_OPENDISTRO_MONITOR_BASE_URI = "/_opendistro/_alerting/monitors"
+        @JvmField val LEGACY_OPENDISTRO_WORKFLOW_BASE_URI = "/_opendistro/_alerting/workflows"


This only for legacy APIs. This is a new API, so we should not have this

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

… checking workflow metadata. Changed flow of workflow execution Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

stevanbz · 2023-03-13T08:53:14Z

can you look into adding this painless script module at plugin load test there will be some method you can override and register the necessary plugin

Let's look into how we can verify composite monitors containing bucket level monitors

…hat verify that workflow metadata is not created Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

…he monitors once the workflow is updated Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

stevanbz added 2 commits January 27, 2023 19:46

Added integrations tests for checking workflow creation and update sc…

56bba0d

…enario Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added transport layer for getting and deleting the workflow

e0af305

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Updated getting and deleting the workflow in order to check if the mo…

feebf0e

…nitor index is not initialized yet. Added workflow crud test cases Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep suggested changes Feb 24, 2023

View reviewed changes

eirsep reviewed Feb 24, 2023

View reviewed changes

stevanbz added 5 commits February 27, 2023 16:24

When deleting the monitor, added a check if the monitor is part of th…

22eb900

…e workflow Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Removed unused classes

c2588a0

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added rest action for executing the workflow

ad01b70

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added integration tests for workflow execution. Added script modules …

5190726

…when loading the cluster Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep suggested changes Feb 27, 2023

View reviewed changes

Added workflow execution run result and refactored ExecutionWorkflowR…

a1e0408

…esponse class Code adjusted according to comments Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

stevanbz force-pushed the feature/composite-workflow-execution-v1 branch from d94c257 to a1e0408 Compare February 27, 2023 22:48

Added integration tests for workflow execution. PR comments addressed

a77d9bb

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

eirsep suggested changes Mar 1, 2023

View reviewed changes

Code adjusted to comments. Wrapped exceptions when executing workflow

b6f17a8

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added logic for deleting the workflow underlying monitors. Added vali…

8ded8b8

…dation if the query monitor is part of the workflow chain Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

lezzago reviewed Mar 8, 2023

View reviewed changes

stevanbz added 2 commits March 9, 2023 18:47

Added workflow metadata

a593d38

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added mappings for the workflow-metadata. Added integration tests for…

8e0d28d

… checking workflow metadata. Changed flow of workflow execution Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

stevanbz force-pushed the feature/composite-workflow-execution-v1 branch from 20de168 to 8e0d28d Compare March 9, 2023 21:17

stevanbz changed the base branch from feature/composite-workflow-v1 to feature/composite-workflow-transport-crud-execution March 9, 2023 22:41

stevanbz closed this Mar 13, 2023

stevanbz reopened this Mar 13, 2023

stevanbz added 2 commits March 13, 2023 11:46

Renamed properties. Added workflow metadata dryrun integration test t…

af86c69

…hat verify that workflow metadata is not created Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added workflow integration test for verifying changing the order of t…

4dd13ed

…he monitors once the workflow is updated Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

stevanbz added 2 commits March 13, 2023 16:10

Renamed methods for generating the workflows

a14dfea

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>

Added test when updating the non-existing workflow

486c5ab

Signed-off-by: Stevan Buzejic <buzejic.stevan@gmail.com>


		class ExecuteWorkflowResponse : ActionResponse, ToXContentObject {

		val workflowRunResult: List<MonitorRunResult<*>>

Conversation

stevanbz commented Feb 13, 2023

Uh oh!

eirsep commented Feb 20, 2023

Uh oh!

eirsep left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevanbz Mar 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevanbz Feb 27, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevanbz Feb 27, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevanbz Feb 27, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevanbz Mar 13, 2023 •

edited

Loading

stevanbz Feb 27, 2023 •

edited

Loading

stevanbz Feb 27, 2023 •

edited

Loading

stevanbz Feb 27, 2023 •

edited

Loading

stevanbz Mar 2, 2023 •

edited

Loading